package com.flink.window;

import com.flink.entity.User;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.time.LocalDate;

/**
 * 描述:
 * 窗口分配器
 *
 * @author yanzhengwu
 * @create 2022-07-28 19:23
 */
public class WindowAssigners {

    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的事100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
        env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.fromElements(
                new User("用户2", "家电", LocalDate.of(2022, 6, 2), 12, 10000L),
                new User("用户1", "零食", LocalDate.of(2022, 1, 1), 11, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户3", "椅子", LocalDate.of(2022, 6, 3), 14, 10000L),
                new User("用户3", "手机", LocalDate.of(2022, 6, 3), 15, 10000L),
                new User("用户1", "电脑", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户1", "手机", LocalDate.of(2022, 6, 16), 16, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 60000L)
        )       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定已事件时间的具体时间戳字段
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));
        //必须先做keyby才能执行开窗，不要问为什么这就是执行机制，非得深究可以去看flink设计思想
        stream.keyBy(User::getName)
                //这里为计数窗口 单个参数为滚动窗口 双参数为滑动窗口，TODO 这里为10个数据统计一次，2秒滑动一次
//                .countWindow(10,2)
                //滚动事件处理窗口，TODO 这里是给了1小时一个窗口
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                //滚动事件处理窗口，TODO 这里是给了1小时一个窗口，第二个参数是偏移量，可以通过这个参数来倒时差，注意看源码
//                .window(TumblingEventTimeWindows.of(Time.hours(1),Time.minutes(-8)))
                //滑动时间处理窗口，TODO 这里是给了1小时一个窗口,5分钟滑动一次，注：滑动窗口也有第三个参数 偏移量！
//                .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
                //事件会话窗口，TODO 这里给的事5秒一个窗口
                .window(EventTimeSessionWindows.withGap(Time.seconds(5)));

        env.execute();
    }
}
